package cn._51doit.day06.homeWork;

import com.google.common.hash.BloomFilter;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

/**
 * @create: 2021-10-22 16:34
 * @author: 今晚打脑斧先森
 * @program: CheckpointAndZuoYe
 * @Description:
 **/
public class CheckpointAndZuoYe {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        //开启checkpoint,默认是无限重启
        env.enableCheckpointing(5000); //5秒进行一次状态的保存

        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);
        SingleOutputStreamOperator<Tuple3<String, String,Integer>> map1 = lines.map(new MapFunction<String, Tuple3<String, String,Integer>>() {
            @Override
            public Tuple3<String, String,Integer> map(String value) throws Exception {
                String[] split = value.split(",");
                String gg = split[0];
                String user = "user:" + split[1]; //不管你输入的姓名是什么,对给你打个标签
                return Tuple3.of(gg,user,1); //元组(广告,用户名)
            }
        });
        //按照广告进行的分区
        KeyedStream<Tuple3<String, String,Integer>, String> keyedStream = map1.keyBy(tp -> tp.f0);
        SingleOutputStreamOperator<Tuple3<String, Integer, Integer>> map2 = keyedStream.map(new MyValueState());
        map2.print();
        env.execute();

    }
    private static class MyValueState extends RichMapFunction<Tuple3<String,String,Integer>,Tuple3<String,Integer,Integer>>{
        //人名
//        ValueState<HashMap<String,HashSet<String>>> userName;
//        MapState<String,String> userName;
        ValueState<HashSet<String>> userName ;
        //人次
        ValueState<Integer> UserCount;
        @Override
        public void open(Configuration parameters) throws Exception {

//            ValueStateDescriptor<HashMap<String,HashSet<String>>> descriptor3 = new ValueStateDescriptor<HashMap<String, HashSet<String>>>
//                    ("UserName",TypeInformation.of(new TypeHint<HashMap<String, HashSet<String>>>() {}));
//            userName = getRuntimeContext().getState(descriptor3);
//            MapStateDescriptor<String,String> descriptor = new MapStateDescriptor<String, String>
//                    ("今晚打脑斧",String.class,String.class);
//            userName=getRuntimeContext().getMapState(descriptor);
            ValueStateDescriptor<HashSet<String>> descriptor = new ValueStateDescriptor<>("人数", TypeInformation.of(new TypeHint<HashSet<String>>() {
            }));
            userName = getRuntimeContext().getState(descriptor);


            //存储次数
            ValueStateDescriptor<Integer> descriptor2 = new ValueStateDescriptor<>("次数",Integer.class);
            UserCount = getRuntimeContext().getState(descriptor2);
        }
        HashMap<String, HashSet<String>> hashName;
        HashSet<String> hashName2 ;

        @Override
        public Tuple3<String, Integer, Integer> map(Tuple3<String, String, Integer> value) throws Exception {
            String gg = value.f0;   //广告
            String currentUserName = value.f1;   //当前用户名
            Integer currentCount = value.f2;    //访问次数


            HashSet<String> historyName = userName.value();
//            //如果没有就添加用户
            if (historyName == null){
                historyName = new HashSet<>();
            }
            historyName.add(currentUserName);
            userName.update(historyName);
            int length = historyName.size();

            //人次
            Integer historyCount = UserCount.value();
            if (historyCount==null){
                historyCount=0;
            }
            int totalCount = currentCount+historyCount; //人次累加
            UserCount.update(totalCount);
            return Tuple3.of(gg,length,totalCount);

        }
    }
}
